package com.zhang.req.order;

import com.zhang.second.utils.OrderEvents;
import com.zhang.second.utils.TxEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 实时对账
 * @author: zhang
 * @date: 2022/2/10 11:06
 */
public class OrderReceiptWithTimerState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //支付流
        SingleOutputStreamOperator<OrderEvents> orderEventsDS = env
                .readTextFile("/Users/apple/IdeaProjects/flink_1.13/src/main/resources/OrderLog.csv")
                .map(new MapFunction<String, OrderEvents>() {
                    @Override
                    public OrderEvents map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new OrderEvents(
                                fields[0], fields[1], fields[2],
                                Long.parseLong(fields[3]) * 1000L
                        );
                    }
                })
                .filter(r -> "pay".equals(r.type))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvents>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvents>() {
                                    @Override
                                    public long extractTimestamp(OrderEvents element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );


        //到账流
        SingleOutputStreamOperator<TxEvent> txEventDS = env
                .readTextFile("/Users/apple/IdeaProjects/flink_1.13/src/main/resources/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new TxEvent(
                                fields[0], fields[1],
                                Long.parseLong(fields[2]) * 1000L
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<TxEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<TxEvent>() {
                                    @Override
                                    public long extractTimestamp(TxEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        //对账
        SingleOutputStreamOperator<Tuple2<OrderEvents, TxEvent>> result = orderEventsDS
                .keyBy(r -> r.txId)
                .connect(txEventDS.keyBy(r -> r.txId))
                .process(new CoProcessFunction<OrderEvents, TxEvent, Tuple2<OrderEvents, TxEvent>>() {

                    private ValueState<OrderEvents> orderEventsValueState;
                    private ValueState<TxEvent> txEventValueState;
                    private ValueState<Long> timerTs;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        orderEventsValueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<OrderEvents>(
                                        "order-state", Types.POJO(OrderEvents.class)
                                )
                        );

                        txEventValueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<TxEvent>(
                                        "receipt-state", Types.POJO(TxEvent.class)
                                )
                        );

                        timerTs = getRuntimeContext().getState(
                                new ValueStateDescriptor<Long>(
                                        "timer-ts", Types.LONG
                                )
                        );
                    }

                    @Override
                    public void processElement1(OrderEvents value, CoProcessFunction<OrderEvents, TxEvent, Tuple2<OrderEvents, TxEvent>>.Context ctx, Collector<Tuple2<OrderEvents, TxEvent>> out) throws Exception {
                        if (txEventValueState.value() == null) {
                            //如果到账数据未到达，把支付状态存下来
                            orderEventsValueState.update(value);
                            //注册定时器等待到账数据10s
                            long tenSecondLater = value.ts + 10 * 1000L;
                            ctx.timerService().registerEventTimeTimer(tenSecondLater);
                            //储存定时器时间状态
                            timerTs.update(tenSecondLater);
                        } else {
                            //当支付数据来的时候，到账数据已经到达，则对账成功正常输出
                            out.collect(Tuple2.of(value, txEventValueState.value()));
                            //删除定时器
                            ctx.timerService().deleteEventTimeTimer(timerTs.value());
                            //清空状态
                            txEventValueState.clear();
                            timerTs.clear();
                        }
                    }

                    @Override
                    public void processElement2(TxEvent value, CoProcessFunction<OrderEvents, TxEvent, Tuple2<OrderEvents, TxEvent>>.Context ctx, Collector<Tuple2<OrderEvents, TxEvent>> out) throws Exception {
                        if (orderEventsValueState.value() == null) {
                            //说明到账数据已经到达倒数支付数据还未到达，把到账数据存下来
                            txEventValueState.update(value);
                            //注册定时器等待支付数据5s
                            long fiveSecondLater = value.ts + 5 * 1000L;
                            ctx.timerService().registerEventTimeTimer(fiveSecondLater);
                            //储存定时器时间状态
                            timerTs.update(fiveSecondLater);
                        } else {
                            //当到账数据来的时候，支付数据已经到达则对账成功，正常输出
                            out.collect(Tuple2.of(orderEventsValueState.value(), value));
                            //删除支付数据先到时候注册的定时器
                            ctx.timerService().deleteEventTimeTimer(timerTs.value());
                            //清空状态
                            orderEventsValueState.clear();
                            timerTs.clear();
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, CoProcessFunction<OrderEvents, TxEvent, Tuple2<OrderEvents, TxEvent>>.OnTimerContext ctx, Collector<Tuple2<OrderEvents, TxEvent>> out) throws Exception {
                        //只要是触发定时器 （ 支付为空，到账不为空；支付不为空，到账为空），所以只对一个状态进行判断即可
                        if (orderEventsValueState.value() != null) {
                            //说明支付数据到达，当定时器触发时候到账数据还未到达
                            ctx.output(
                                    new OutputTag<String>("no-tx") {
                                    },
                                    orderEventsValueState.value().txId + "只有支付数据没有到账数据"
                            );
                        } else {
                            //说明没有支付数据只有到账数据
                            ctx.output(
                                    new OutputTag<String>("no-order") {
                                    },
                                    txEventValueState.value().txId + "只有到账数据没有支付数据"
                            );
                        }
                    }
                });
        result.print();
        result.getSideOutput(new OutputTag<String>("no-tx") {
        }).print("no-tx");
        result.getSideOutput( new OutputTag<String>("no-order") {
        }).print("no-order");


        env.execute();
    }
}
